package com.fy.service.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.fy.common.constant.CmdConstants;
import com.fy.common.constant.Constants;
import com.fy.common.constant.RedisKeyConstant;
import com.fy.common.entity.SkImChatSession;
import com.fy.common.entity.SkImUserInfo;
import com.fy.common.enums.SkImChatSessionEnum;
import com.fy.common.enums.SkImCommonStatusEnum;
import com.fy.common.model.im.MessageModel;
import com.fy.common.model.im.UserOnlineStatus;
import com.fy.common.util.ImUtil;
import com.fy.dao.mapper.SkImChatSessionMapper;
import com.fy.service.service.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;

@Service
public class ImBrokerServiceImpl implements ImBrokerService {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private RocketMQTemplate rocketMQTemplate;
    private RedissonClient redisson;
    private ImBrokerGroupService imBrokerGroupService;
    private SkImChatSessionService skImChatSessionService;
    private SkImChatSessionMapper skImChatSessionMapper;
    private MessageService messageService;
    private SkImUserInfoService userInfoService;

    @Autowired
    public void setRocketMQTemplate(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }
    @Autowired
    public void setRedisson(RedissonClient redisson) {
        this.redisson = redisson;
    }
    @Autowired
    public void setImBrokerGroupService(ImBrokerGroupService imBrokerGroupService) {
        this.imBrokerGroupService = imBrokerGroupService;
    }
    @Autowired
    public void setSkImChatSessionService(SkImChatSessionService skImChatSessionService) {
        this.skImChatSessionService = skImChatSessionService;
    }
    @Autowired
    public void setSkImChatSessionMapper(SkImChatSessionMapper skImChatSessionMapper) {
        this.skImChatSessionMapper = skImChatSessionMapper;
    }
    @Autowired
    public void setMessageService(MessageService messageService) {
        this.messageService = messageService;
    }
    @Autowired
    public void setUserInfoService(SkImUserInfoService userInfoService) {
        this.userInfoService = userInfoService;
    }

    @Override
    public void sendSingleMessage(MessageModel messageModel) {

        //获取用户每台设备的netty host地址
        List<UserOnlineStatus> onlineList = this.getAllPushDevices(messageModel);
        for (UserOnlineStatus onlineStatus : onlineList) {
            //对下发节点推送通知  设置要推送的设备code
            messageModel.setDeviceCode(onlineStatus.getDeviceCode());
            rocketMQTemplate.convertAndSend(onlineStatus.getLastOnlineHost().replace(".", "_").replace(":", "_"), JSONObject.toJSONString(messageModel));
        }
    }

    @Override
    public List<UserOnlineStatus> getAllPushDevices(MessageModel messageModel) {
        List list = this.getFromPushDevices(messageModel);
        list.addAll(this.getToPushDevices(messageModel));

        return list;
    }

    @Override
    public List<UserOnlineStatus> getFromPushDevices(MessageModel messageModel) {
        List<UserOnlineStatus> userOnlineStatuses = new ArrayList<>();

        //发送方的所有待推送设备
        RMapCache<String, JSONObject> fromDevices = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + messageModel.getFrom());
        //移除当前发送设备 不做推送
        //fromDevices.remove(messageModel.getDeviceCode());
        for (String fromDeviceCode : fromDevices.keySet()) {
            UserOnlineStatus onlineStatus = JSONObject.toJavaObject((JSONObject) fromDevices.get(fromDeviceCode).get("UserOnlineStatus"), UserOnlineStatus.class);


            if (onlineStatus != null && onlineStatus.getIsOnline() == Constants.YES) {
                //不在对发送设备进行推送
                if (!onlineStatus.getDeviceCode().equals(messageModel.getDeviceCode()))
                    userOnlineStatuses.add(onlineStatus);
            }
        }
        return userOnlineStatuses;
    }

    @Override
    public List<UserOnlineStatus> getToPushDevices(MessageModel messageModel) {
        List<UserOnlineStatus> userOnlineStatuses = new ArrayList<>();
        //接收方的所有待推送设备
        RMapCache<String, JSONObject> toDevices = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + messageModel.getTo());
        //获取接受方的所有设备
        for (String toDeviceCode : toDevices.keySet()) {
            UserOnlineStatus onlineStatus = JSONObject.toJavaObject((JSONObject) toDevices.get(toDeviceCode).get("UserOnlineStatus"), UserOnlineStatus.class);
            if (onlineStatus != null && onlineStatus.getIsOnline() == Constants.YES) {
                userOnlineStatuses.add(onlineStatus);
            }
        }
        return userOnlineStatuses;
    }


    @Override
    public void updateToSessionLastMessage(MessageModel messageModel, SkImChatSession session) {
        RMapCache<String, SkImChatSession> toSession = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.USER_CHAT_SESSION + messageModel.getTo());
        String key = null;
        if (messageModel.getToGroup() == null) {
            key = ImUtil.chatSessionIdGenFriend(messageModel.getTo(), messageModel.getFrom());
        } else {
            key = ImUtil.chatSessionIdGenGroup(messageModel.getTo(), messageModel.getToGroup());
        }
        if (session == null) {
            session = toSession.get(key);
            if (session == null) {
                session = skImChatSessionService.selectOne(new EntityWrapper<SkImChatSession>().eq("session_id", key));
                SkImUserInfo imUserInfo = userInfoService.selectOne(new EntityWrapper<SkImUserInfo>().eq("sk_im_user_id", session.getTargetId()));
                if (Objects.nonNull(imUserInfo)) {
                    session.setAvatar(imUserInfo.getAvatar()).setChatTitle(imUserInfo.getUsername());
                }
                if (session == null) throw new RuntimeException("找不到会话");
            }


        } else {
            key = session.getSessionId();
        }
        //恢复被删除的会话
        this.recoverSession(session);
        session.setContentType(messageModel.getContentType());
        session.setCmd(messageModel.getCmd());
        session.setContent(messageModel.getContent());
        session.setLastMsgTime(new Date());
        session.setUnreadCount(session.getUnreadCount() + 1);
        toSession.putAsync(key, session);

    }



    @Override
    public void updateFromSessionLastMessage(MessageModel messageModel, SkImChatSession session) {
        RMapCache<String, SkImChatSession> fromSession = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.USER_CHAT_SESSION + messageModel.getFrom());
        String key = null;
        if (messageModel.getToGroup() == null) {
            key = ImUtil.chatSessionIdGenFriend(messageModel.getFrom(), messageModel.getTo());
        } else {
            key = ImUtil.chatSessionIdGenGroup(messageModel.getFrom(), messageModel.getToGroup());
        }
        if (session == null) {
            session = fromSession.get(key);
            if (session == null) {
                session = skImChatSessionService.selectOne(new EntityWrapper<SkImChatSession>().eq("session_id", key));
                SkImUserInfo imUserInfo = userInfoService.selectOne(new EntityWrapper<SkImUserInfo>().eq("sk_im_user_id", session.getTargetId()));
                if (Objects.nonNull(imUserInfo)) {
                    session.setAvatar(imUserInfo.getAvatar()).setChatTitle(imUserInfo.getUsername());
                }
                if (session == null) throw new RuntimeException("找不到会话");
            }
        } else {
            key = session.getSessionId();
        }
        //恢复被删除的会话
        this.recoverSession(session);

        //对发送方发起 黑名单之类的权限验证
        if (session.getStatus() == SkImCommonStatusEnum.BLACK_DISABLE.getStatus()) {
            MessageModel responseModel = new MessageModel();
            responseModel.setCmd(CmdConstants.RESPONSE_SEND_SUCCESS);
            responseModel.setToken(messageModel.getToken());
            responseModel.setDeviceCode(messageModel.getDeviceCode());
            sendToMyOtherDevice(responseModel);
            return;
        }
        session.setContentType(messageModel.getContentType());
        session.setCmd(messageModel.getCmd());
        session.setContent(messageModel.getContent());
        session.setLastMsgTime(new Date());
        fromSession.putAsync(key, session);
    }

    @Override
    public void updateSingleChatSession(MessageModel messageModel) {

        //修改自己的会话
        this.updateFromSessionLastMessage(messageModel, null);
        this.updateToSessionLastMessage(messageModel, null);

    }

    @Override
    public void sendTyping(MessageModel messageModel) {
        List<UserOnlineStatus> onlineStatusList = this.getToPushDevices(messageModel);
        for (UserOnlineStatus onlineStatus : onlineStatusList) {
            //对下发节点推送通知  设置要推送的设备code
            messageModel.setDeviceCode(onlineStatus.getDeviceCode());
            rocketMQTemplate.convertAndSend(onlineStatus.getLastOnlineHost().replace(".", "_").replace(":", "_"), JSONObject.toJSONString(messageModel));
        }

    }

    @Override
    public void push(MessageModel message, ChannelHandlerContext ctx) throws Exception {
        if (message.getProtocolType() == Constants.WEB_SOCKET) {
            ctx.write(new TextWebSocketFrame(JSONObject.toJSONString(message)));
        } else {
            ctx.write(JSONObject.toJSONString(message));
        }
    }

    @Override
    public void push(MessageModel message, Channel ctx) throws Exception {
        if (message.getProtocolType() == Constants.WEB_SOCKET) {
            ctx.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(message)));
        } else {
            ctx.writeAndFlush(JSONObject.toJSONString(message));
        }
    }


    @Override
    public void commonAddressingMessage(MessageModel messageModel, List<Long> targets, Boolean isSendMyself) {
        List<UserOnlineStatus> onlineList = new ArrayList<>();

        if (isSendMyself) {
            onlineList.addAll(this.getFromPushDevices(messageModel));
        }

        for (Long target : targets) {
            //获取用户每台设备的netty host地址
            messageModel.setTo(target);
            onlineList.addAll(this.getToPushDevices(messageModel));
            for (UserOnlineStatus onlineStatus : onlineList) {
                //对下发节点推送通知  设置要推送的设备code
                messageModel.setDeviceCode(onlineStatus.getDeviceCode());
                rocketMQTemplate.convertAndSend(onlineStatus.getLastOnlineHost().replace(".", "_").replace(":", "_"), JSONObject.toJSONString(messageModel));
            }
        }

    }

    @Override
    public void sendToMyOtherDevice(MessageModel messageModel) {
        List<UserOnlineStatus> onlineList = this.getFromPushDevices(messageModel);
        for (UserOnlineStatus onlineStatus : onlineList) {
            //对下发节点推送通知  设置要推送的设备code
            messageModel.setDeviceCode(onlineStatus.getDeviceCode());
            rocketMQTemplate.convertAndSend(onlineStatus.getLastOnlineHost().replace(".", "_").replace(":", "_"), JSONObject.toJSONString(messageModel));
        }
    }

    @Override
    public void sendRecall(MessageModel messageModel) {

        if (messageModel.getToGroup() == null) {
            //重用单聊的下发业务
            this.sendSingleMessage(messageModel);
        } else {
            imBrokerGroupService.sendGroupMessage(messageModel, false);
        }
    }

    @Override
    public void readMessage(MessageModel messageModel) {
        //修改会话缓存中的小红点
        RMapCache<String, SkImChatSession> fromSession = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.USER_CHAT_SESSION + messageModel.getFrom());

        String key = "";
        if (messageModel.getToGroup() == null) {
            key = ImUtil.chatSessionIdGenFriend(messageModel.getFrom(), messageModel.getTo());
        } else {
            key = ImUtil.chatSessionIdGenGroup(messageModel.getFrom(), messageModel.getToGroup());
        }

        SkImChatSession session = fromSession.get(key);
        if (session != null) {
            session.setUnreadCount(0);
            fromSession.put(key, session);
        }


        if (messageModel.getToGroup() != null) {
            this.sendToMyOtherDevice(messageModel);
        } else {
            //通知对方和自己其他设备消息已读
            List<Long> to = new ArrayList<>();
            to.add(messageModel.getTo());
            this.commonAddressingMessage(messageModel, to, true);
            //写入已读的离线消息到hbase
            MessageModel model = new MessageModel();
            BeanUtils.copyProperties(messageModel, model);
            model.setCmd(messageModel.getCmd());
            messageService.saveToMessage(model);
        }

    }

    @Override
    public void recoverSession(SkImChatSession session) {
        if (session.getLastDeleteType() == SkImChatSessionEnum.DEL.getType()) {
            session.setLastDeleteType(SkImChatSessionEnum.NORMAL.getType());
            skImChatSessionMapper.recoverSession(session.getId(), SkImChatSessionEnum.NORMAL.getType());
        }
    }

    @Override
    public List<UserOnlineStatus> getOnlineDevice(String tenantId, Long userId) {
        RMapCache<String, JSONObject> toDevices = redisson.getMapCache(tenantId + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + userId);
        List<UserOnlineStatus> onlineList = new ArrayList<>();
        //获取接受方的所有设备
        for (String toDeviceCode : toDevices.keySet()) {
            UserOnlineStatus onlineStatus = JSONObject.toJavaObject((JSONObject) toDevices.get(toDeviceCode).get("UserOnlineStatus"), UserOnlineStatus.class);
            if (onlineStatus != null && onlineStatus.getIsOnline() == Constants.YES) {
                onlineList.add(onlineStatus);
            }
        }
        return onlineList;
    }


}
